package com.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

import java.io.File

object Demo1WordCount {

  def main(args: Array[String]): Unit = {
    /**
     * 1、创建spark环境
     */
    //spark环境配置对象
    val conf = new SparkConf()

    //设置spark任务的名称
    conf.setAppName("wordcount")
    //设置spark运行模式，local:本地运行
    conf.setMaster("local")

    //创建spark上下文对象，sc是spark写代码的入口
    val sc = new SparkContext(conf)

    // 日志级别
    sc.setLogLevel("error")

    println(new File(".").getAbsoluteFile)

    /**
     * 2、读取文件
     * spark读取文件底层的代码和MapReduce是一样的
     * 所以切片规则一样
     * spark是一个切片对应一个分区
     *
     * RDD：弹性的分布式数据集，现阶段可以当初list集合使用
     */
    val linesRDD: RDD[String] = sc.textFile("data/words.txt")

    /**
     * 3、将单词展开
     */
    val wordsRdd: RDD[String] = linesRDD.flatMap(line => line.split(","))

    /**
     * 4、按照单词分组
     */
    val kvRdd: RDD[(String, Iterable[String])] = wordsRdd.groupBy(word => word)

    /**
     * 5、统计单词的数量
     */
    val wordCount: RDD[String] = kvRdd.map {
      case (word: String, iter: Iterable[String]) =>
        val count: Int = iter.size
        s"$word\t$count"
    }

    wordCount.foreach(println)

    /**
     * 6、保存数据
     */
    wordCount.saveAsTextFile("data/wordcount")

  }
}
